/*-------------------------------------------------------------------------
 *
 * hash_xlog.c
 *	  哈希索引的WAL重放逻辑。
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/access/hash/hash_xlog.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/bufmask.h"
#include "access/hash.h"
#include "access/hash_xlog.h"
#include "access/transam.h"
#include "access/xlog.h"
#include "access/xlogutils.h"
#include "miscadmin.h"
#include "storage/procarray.h"

/*
 * 重放哈希索引元页面
 */
static void fc_hash_xlog_init_meta_page(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	Page		fc_page;
	Buffer		fc_metabuf;
	ForkNumber	fc_forknum;

	xl_hash_init_meta_page *fc_xlrec = (xl_hash_init_meta_page *) XLogRecGetData(fc_record);

	/* 创建索引的元页面 */
	fc_metabuf = XLogInitBufferForRedo(fc_record, 0);
	Assert(BufferIsValid(fc_metabuf));
	_hash_init_metabuffer(fc_metabuf, fc_xlrec->num_tuples, fc_xlrec->procid,
						  fc_xlrec->ffactor, true);
	fc_page = (Page) BufferGetPage(fc_metabuf);
	PageSetLSN(fc_page, fc_lsn);
	MarkBufferDirty(fc_metabuf);

	/*
	 * 强制初始化分支的磁盘状态始终与共享缓冲区中的状态保持同步。
	 * 参见 XLogReadBufferForRedoExtended。我们需要对初始化分支进行特殊处理，
	 * 因为创建索引操作不会记录元页面的完整页面图像。
	 */
	XLogRecGetBlockTag(fc_record, 0, NULL, &fc_forknum, NULL);
	if (fc_forknum == INIT_FORKNUM)
		FlushOneBuffer(fc_metabuf);

	/* 全部完成 */
	UnlockReleaseBuffer(fc_metabuf);
}

/*
 * 重放哈希索引位图页面
 */
static void fc_hash_xlog_init_bitmap_page(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	Buffer		fc_bitmapbuf;
	Buffer		fc_metabuf;
	Page		fc_page;
	HashMetaPage fc_metap;
	uint32		fc_num_buckets;
	ForkNumber	fc_forknum;

	xl_hash_init_bitmap_page *fc_xlrec = (xl_hash_init_bitmap_page *) XLogRecGetData(fc_record);

	/*
	 * 初始化位图页面
	 */
	fc_bitmapbuf = XLogInitBufferForRedo(fc_record, 0);
	_hash_initbitmapbuffer(fc_bitmapbuf, fc_xlrec->bmsize, true);
	PageSetLSN(BufferGetPage(fc_bitmapbuf), fc_lsn);
	MarkBufferDirty(fc_bitmapbuf);

	/*
	 * 强制初始化分支的磁盘状态始终与共享缓冲区中的状态保持同步。
	 * 参见 XLogReadBufferForRedoExtended。我们需要对初始化分支进行特殊处理，
	 * 因为创建索引操作不会记录元页面的完整页面图像。
	 */
	XLogRecGetBlockTag(fc_record, 0, NULL, &fc_forknum, NULL);
	if (fc_forknum == INIT_FORKNUM)
		FlushOneBuffer(fc_bitmapbuf);
	UnlockReleaseBuffer(fc_bitmapbuf);

	/* 将新位图页面添加到元页面的位图列表中 */
	if (XLogReadBufferForRedo(fc_record, 1, &fc_metabuf) == BLK_NEEDS_REDO)
	{
		/*
		 * 注意：在正常操作中，我们会在仍持有位图页面的锁时更新元页面。
		 * 但在重放期间不需要持有该锁，因为还没有人能看到它；创建事务尚未提交。
		 */
		fc_page = BufferGetPage(fc_metabuf);
		fc_metap = HashPageGetMeta(fc_page);

		fc_num_buckets = fc_metap->hashm_maxbucket + 1;
		fc_metap->hashm_mapp[fc_metap->hashm_nmaps] = fc_num_buckets + 1;
		fc_metap->hashm_nmaps++;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_metabuf);

		XLogRecGetBlockTag(fc_record, 1, NULL, &fc_forknum, NULL);
		if (fc_forknum == INIT_FORKNUM)
			FlushOneBuffer(fc_metabuf);
	}
	if (BufferIsValid(fc_metabuf))
		UnlockReleaseBuffer(fc_metabuf);
}

/*
 * 重放哈希索引插入（不分裂）
 */
static void fc_hash_xlog_insert(XLogReaderState *fc_record)
{
	HashMetaPage fc_metap;
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_insert *fc_xlrec = (xl_hash_insert *) XLogRecGetData(fc_record);
	Buffer		fc_buffer;
	Page		fc_page;

	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		Size		fc_datalen;
		char	   *fc_datapos = XLogRecGetBlockData(fc_record, 0, &fc_datalen);

		fc_page = BufferGetPage(fc_buffer);

		if (PageAddItem(fc_page, (Item) fc_datapos, fc_datalen, fc_xlrec->offnum,
						false, false) == InvalidOffsetNumber)
			elog(PANIC, "hash_xlog_insert: failed to add item");

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	if (XLogReadBufferForRedo(fc_record, 1, &fc_buffer) == BLK_NEEDS_REDO)
	{
		/*
		 * 注意：在正常操作中，我们会在仍持有插入页面的锁时更新元页面。
		 * 但在重放期间不需要持有该锁，因为不会有其他索引更新同时发生。
		 */
		fc_page = BufferGetPage(fc_buffer);
		fc_metap = HashPageGetMeta(fc_page);
		fc_metap->hashm_ntuples += 1;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

/*
 * 重放为哈希索引添加溢出页面
 */
static void fc_hash_xlog_add_ovfl_page(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_add_ovfl_page *fc_xlrec = (xl_hash_add_ovfl_page *) XLogRecGetData(fc_record);
	Buffer		fc_leftbuf;
	Buffer		fc_ovflbuf;
	Buffer		fc_metabuf;
	BlockNumber fc_leftblk;
	BlockNumber fc_rightblk;
	BlockNumber fc_newmapblk = InvalidBlockNumber;
	Page		fc_ovflpage;
	HashPageOpaque fc_ovflopaque;
	uint32	   *fc_num_bucket;
	char	   *fc_data;
	Size		fc_datalen PG_USED_FOR_ASSERTS_ONLY;
	bool		fc_new_bmpage = false;

	XLogRecGetBlockTag(fc_record, 0, NULL, NULL, &fc_rightblk);
	XLogRecGetBlockTag(fc_record, 1, NULL, NULL, &fc_leftblk);

	fc_ovflbuf = XLogInitBufferForRedo(fc_record, 0);
	Assert(BufferIsValid(fc_ovflbuf));

	fc_data = XLogRecGetBlockData(fc_record, 0, &fc_datalen);
	fc_num_bucket = (uint32 *) fc_data;
	Assert(fc_datalen == sizeof(uint32));
	_hash_initbuf(fc_ovflbuf, InvalidBlockNumber, *fc_num_bucket, LH_OVERFLOW_PAGE,
				  true);
	/* 更新反向链接 */
	fc_ovflpage = BufferGetPage(fc_ovflbuf);
	fc_ovflopaque = HashPageGetOpaque(fc_ovflpage);
	fc_ovflopaque->hasho_prevblkno = fc_leftblk;

	PageSetLSN(fc_ovflpage, fc_lsn);
	MarkBufferDirty(fc_ovflbuf);

	if (XLogReadBufferForRedo(fc_record, 1, &fc_leftbuf) == BLK_NEEDS_REDO)
	{
		Page		fc_leftpage;
		HashPageOpaque fc_leftopaque;

		fc_leftpage = BufferGetPage(fc_leftbuf);
		fc_leftopaque = HashPageGetOpaque(fc_leftpage);
		fc_leftopaque->hasho_nextblkno = fc_rightblk;

		PageSetLSN(fc_leftpage, fc_lsn);
		MarkBufferDirty(fc_leftbuf);
	}

	if (BufferIsValid(fc_leftbuf))
		UnlockReleaseBuffer(fc_leftbuf);
	UnlockReleaseBuffer(fc_ovflbuf);

	/*
	 * 注意：在正常操作中，我们会在仍持有溢出页面的锁时更新位图和元页面。
	 * 但在重放期间不需要持有这些锁，因为不会有其他索引更新同时发生。
	 */
	if (XLogRecHasBlockRef(fc_record, 2))
	{
		Buffer		fc_mapbuffer;

		if (XLogReadBufferForRedo(fc_record, 2, &fc_mapbuffer) == BLK_NEEDS_REDO)
		{
			Page		fc_mappage = (Page) BufferGetPage(fc_mapbuffer);
			uint32	   *fc_freep = NULL;
			char	   *fc_data;
			uint32	   *fc_bitmap_page_bit;

			fc_freep = HashPageGetBitmap(fc_mappage);

			fc_data = XLogRecGetBlockData(fc_record, 2, &fc_datalen);
			fc_bitmap_page_bit = (uint32 *) fc_data;

			SETBIT(fc_freep, *fc_bitmap_page_bit);

			PageSetLSN(fc_mappage, fc_lsn);
			MarkBufferDirty(fc_mapbuffer);
		}
		if (BufferIsValid(fc_mapbuffer))
			UnlockReleaseBuffer(fc_mapbuffer);
	}

	if (XLogRecHasBlockRef(fc_record, 3))
	{
		Buffer		fc_newmapbuf;

		fc_newmapbuf = XLogInitBufferForRedo(fc_record, 3);

		_hash_initbitmapbuffer(fc_newmapbuf, fc_xlrec->bmsize, true);

		fc_new_bmpage = true;
		fc_newmapblk = BufferGetBlockNumber(fc_newmapbuf);

		MarkBufferDirty(fc_newmapbuf);
		PageSetLSN(BufferGetPage(fc_newmapbuf), fc_lsn);

		UnlockReleaseBuffer(fc_newmapbuf);
	}

	if (XLogReadBufferForRedo(fc_record, 4, &fc_metabuf) == BLK_NEEDS_REDO)
	{
		HashMetaPage fc_metap;
		Page		fc_page;
		uint32	   *fc_firstfree_ovflpage;

		fc_data = XLogRecGetBlockData(fc_record, 4, &fc_datalen);
		fc_firstfree_ovflpage = (uint32 *) fc_data;

		fc_page = BufferGetPage(fc_metabuf);
		fc_metap = HashPageGetMeta(fc_page);
		fc_metap->hashm_firstfree = *fc_firstfree_ovflpage;

		if (!fc_xlrec->bmpage_found)
		{
			fc_metap->hashm_spares[fc_metap->hashm_ovflpoint]++;

			if (fc_new_bmpage)
			{
				Assert(BlockNumberIsValid(fc_newmapblk));

				fc_metap->hashm_mapp[fc_metap->hashm_nmaps] = fc_newmapblk;
				fc_metap->hashm_nmaps++;
				fc_metap->hashm_spares[fc_metap->hashm_ovflpoint]++;
			}
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_metabuf);
	}
	if (BufferIsValid(fc_metabuf))
		UnlockReleaseBuffer(fc_metabuf);
}

/*
 * 重放分裂操作的页面分配
 */
static void fc_hash_xlog_split_allocate_page(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_split_allocate_page *fc_xlrec = (xl_hash_split_allocate_page *) XLogRecGetData(fc_record);
	Buffer		fc_oldbuf;
	Buffer		fc_newbuf;
	Buffer		fc_metabuf;
	Size		fc_datalen PG_USED_FOR_ASSERTS_ONLY;
	char	   *fc_data;
	XLogRedoAction fc_action;

	/*
	 * 为了与正常操作保持一致，此处我们在旧存储桶和新存储桶上都采取清理锁，
	 * 尽管不会有并发插入。
	 */

	/* 重放旧存储桶的记录 */
	fc_action = XLogReadBufferForRedoExtended(fc_record, 0, RBM_NORMAL, true, &fc_oldbuf);

	/*
	 * 请注意，即使页面是从完整页面图像恢复的，我们仍然会更新页面，
	 * 因为特定空间不包含在图像中。
	 */
	if (fc_action == BLK_NEEDS_REDO || fc_action == BLK_RESTORED)
	{
		Page		fc_oldpage;
		HashPageOpaque fc_oldopaque;

		fc_oldpage = BufferGetPage(fc_oldbuf);
		fc_oldopaque = HashPageGetOpaque(fc_oldpage);

		fc_oldopaque->hasho_flag = fc_xlrec->old_bucket_flag;
		fc_oldopaque->hasho_prevblkno = fc_xlrec->new_bucket;

		PageSetLSN(fc_oldpage, fc_lsn);
		MarkBufferDirty(fc_oldbuf);
	}

	/* 重放新存储桶的记录 */
	XLogReadBufferForRedoExtended(fc_record, 1, RBM_ZERO_AND_CLEANUP_LOCK, true,
								  &fc_newbuf);
	_hash_initbuf(fc_newbuf, fc_xlrec->new_bucket, fc_xlrec->new_bucket,
				  fc_xlrec->new_bucket_flag, true);
	MarkBufferDirty(fc_newbuf);
	PageSetLSN(BufferGetPage(fc_newbuf), fc_lsn);

	/*
	 * 我们可以提前释放旧存储桶的锁，但在这里这样做是为了与正常操作保持一致。
	 */
	if (BufferIsValid(fc_oldbuf))
		UnlockReleaseBuffer(fc_oldbuf);
	if (BufferIsValid(fc_newbuf))
		UnlockReleaseBuffer(fc_newbuf);

	/*
	 * 注意：在正常操作中，我们会在仍持有旧存储桶和新存储桶页面的锁时更新元页面。
	 * 但在重放期间不需要持有这些锁，因为不会有其他存储桶分裂同时发生。
	 */

	/* 重放元页面更改的记录 */
	if (XLogReadBufferForRedo(fc_record, 2, &fc_metabuf) == BLK_NEEDS_REDO)
	{
		Page		fc_page;
		HashMetaPage fc_metap;

		fc_page = BufferGetPage(fc_metabuf);
		fc_metap = HashPageGetMeta(fc_page);
		fc_metap->hashm_maxbucket = fc_xlrec->new_bucket;

		fc_data = XLogRecGetBlockData(fc_record, 2, &fc_datalen);

		if (fc_xlrec->flags & XLH_SPLIT_META_UPDATE_MASKS)
		{
			uint32		fc_lowmask;
			uint32	   *fc_highmask;

			/* 提取低位和高位掩码。 */
			memcpy(&fc_lowmask, fc_data, sizeof(uint32));
			fc_highmask = (uint32 *) ((char *) fc_data + sizeof(uint32));

			/* 更新元页面 */
			fc_metap->hashm_lowmask = fc_lowmask;
			fc_metap->hashm_highmask = *fc_highmask;

			fc_data += sizeof(uint32) * 2;
		}

		if (fc_xlrec->flags & XLH_SPLIT_META_UPDATE_SPLITPOINT)
		{
			uint32		fc_ovflpoint;
			uint32	   *fc_ovflpages;

			/* 提取溢出页面的信息。 */
			memcpy(&fc_ovflpoint, fc_data, sizeof(uint32));
			fc_ovflpages = (uint32 *) ((char *) fc_data + sizeof(uint32));

			/* 更新元页面 */
			fc_metap->hashm_spares[fc_ovflpoint] = *fc_ovflpages;
			fc_metap->hashm_ovflpoint = fc_ovflpoint;
		}

		MarkBufferDirty(fc_metabuf);
		PageSetLSN(BufferGetPage(fc_metabuf), fc_lsn);
	}

	if (BufferIsValid(fc_metabuf))
		UnlockReleaseBuffer(fc_metabuf);
}

/*
 * 重放分裂操作
 */
static void fc_hash_xlog_split_page(XLogReaderState *fc_record)
{
	Buffer		fc_buf;

	if (XLogReadBufferForRedo(fc_record, 0, &fc_buf) != BLK_RESTORED)
		elog(ERROR, "Hash split record did not contain a full-page image");

	UnlockReleaseBuffer(fc_buf);
}

/*
 * 重放分裂操作完成
 */
static void fc_hash_xlog_split_complete(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_split_complete *fc_xlrec = (xl_hash_split_complete *) XLogRecGetData(fc_record);
	Buffer		fc_oldbuf;
	Buffer		fc_newbuf;
	XLogRedoAction fc_action;

	/* 重放旧存储桶的记录 */
	fc_action = XLogReadBufferForRedo(fc_record, 0, &fc_oldbuf);

	/*
	 * 注意，即使页面是从完整页面图像恢复的，我们仍然会更新页面，
	 * 因为桶标志未包含在图像中。
	 */
	if (fc_action == BLK_NEEDS_REDO || fc_action == BLK_RESTORED)
	{
		Page		fc_oldpage;
		HashPageOpaque fc_oldopaque;

		fc_oldpage = BufferGetPage(fc_oldbuf);
		fc_oldopaque = HashPageGetOpaque(fc_oldpage);

		fc_oldopaque->hasho_flag = fc_xlrec->old_bucket_flag;

		PageSetLSN(fc_oldpage, fc_lsn);
		MarkBufferDirty(fc_oldbuf);
	}
	if (BufferIsValid(fc_oldbuf))
		UnlockReleaseBuffer(fc_oldbuf);

	/* 重放新存储桶的记录 */
	fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_newbuf);

	/*
	 * 注意，即使页面是从完整页面图像恢复的，我们仍然会更新页面，
	 * 因为桶标志未包含在图像中。
	 */
	if (fc_action == BLK_NEEDS_REDO || fc_action == BLK_RESTORED)
	{
		Page		fc_newpage;
		HashPageOpaque fc_nopaque;

		fc_newpage = BufferGetPage(fc_newbuf);
		fc_nopaque = HashPageGetOpaque(fc_newpage);

		fc_nopaque->hasho_flag = fc_xlrec->new_bucket_flag;

		PageSetLSN(fc_newpage, fc_lsn);
		MarkBufferDirty(fc_newbuf);
	}
	if (BufferIsValid(fc_newbuf))
		UnlockReleaseBuffer(fc_newbuf);
}

/*
 * 重放哈希索引的页面内容移动以进行挤压操作
 */
static void fc_hash_xlog_move_page_contents(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_move_page_contents *fc_xldata = (xl_hash_move_page_contents *) XLogRecGetData(fc_record);
	Buffer		fc_bucketbuf = InvalidBuffer;
	Buffer		fc_writebuf = InvalidBuffer;
	Buffer		fc_deletebuf = InvalidBuffer;
	XLogRedoAction fc_action;

	/*
	 * 在开始实际重放操作之前，确保我们在主桶页面上有一个清理锁。
	 * 这样可以确保在此操作的重放期间既没有扫描可以开始，也没有扫描正在进行。
	 * 如果我们在此操作期间允许扫描，则它们可能会漏掉某些记录，
	 * 或者重复显示同一条记录。
	 */
	if (fc_xldata->is_prim_bucket_same_wrt)
		fc_action = XLogReadBufferForRedoExtended(fc_record, 1, RBM_NORMAL, true, &fc_writebuf);
	else
	{
		/*
		 * 我们不在乎返回值，因为读取bucketbuf的目的是确保在主桶页面上有一个清理锁。
		 */
		(void) XLogReadBufferForRedoExtended(fc_record, 0, RBM_NORMAL, true, &fc_bucketbuf);

		fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_writebuf);
	}

	/* 重放用于在溢出缓冲区中添加条目的记录 */
	if (fc_action == BLK_NEEDS_REDO)
	{
		Page		fc_writepage;
		char	   *fc_begin;
		char	   *fc_data;
		Size		fc_datalen;
		uint16		fc_ninserted = 0;

		fc_data = fc_begin = XLogRecGetBlockData(fc_record, 1, &fc_datalen);

		fc_writepage = (Page) BufferGetPage(fc_writebuf);

		if (fc_xldata->ntups > 0)
		{
			OffsetNumber *fc_towrite = (OffsetNumber *) fc_data;

			fc_data += sizeof(OffsetNumber) * fc_xldata->ntups;

			while (fc_data - fc_begin < fc_datalen)
			{
				IndexTuple	fc_itup = (IndexTuple) fc_data;
				Size		fc_itemsz;
				OffsetNumber fc_l;

				fc_itemsz = IndexTupleSize(fc_itup);
				fc_itemsz = MAXALIGN(fc_itemsz);

				fc_data += fc_itemsz;

				fc_l = PageAddItem(fc_writepage, (Item) fc_itup, fc_itemsz, fc_towrite[fc_ninserted], false, false);
				if (fc_l == InvalidOffsetNumber)
					elog(ERROR, "hash_xlog_move_page_contents: failed to add item to hash index page, size %d bytes",
						 (int) fc_itemsz);

				fc_ninserted++;
			}
		}

		/*
		 * 插入的元组数量必须与REDO记录中请求的相同。
		 */
		Assert(fc_ninserted == fc_xldata->ntups);

		PageSetLSN(fc_writepage, fc_lsn);
		MarkBufferDirty(fc_writebuf);
	}

	/* 重放用于从溢出缓冲区中删除条目的记录 */
	if (XLogReadBufferForRedo(fc_record, 2, &fc_deletebuf) == BLK_NEEDS_REDO)
	{
		Page		fc_page;
		char	   *fc_ptr;
		Size		fc_len;

		fc_ptr = XLogRecGetBlockData(fc_record, 2, &fc_len);

		fc_page = (Page) BufferGetPage(fc_deletebuf);

		if (fc_len > 0)
		{
			OffsetNumber *fc_unused;
			OffsetNumber *fc_unend;

			fc_unused = (OffsetNumber *) fc_ptr;
			fc_unend = (OffsetNumber *) ((char *) fc_ptr + fc_len);

			if ((fc_unend - fc_unused) > 0)
				PageIndexMultiDelete(fc_page, fc_unused, fc_unend - fc_unused);
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_deletebuf);
	}

	/*
	 * 重放完成，现在我们可以释放缓冲区。我们在重放操作结束时释放锁，
	 * 以确保在操作结束前我们保持对主桶页面的锁定。
	 * 如果写缓冲区的操作与主桶页面不同，则我们可以优化，
	 * 在相同操作完成后立即释放锁，但这似乎不值得使代码复杂化。
	 */
	if (BufferIsValid(fc_deletebuf))
		UnlockReleaseBuffer(fc_deletebuf);

	if (BufferIsValid(fc_writebuf))
		UnlockReleaseBuffer(fc_writebuf);

	if (BufferIsValid(fc_bucketbuf))
		UnlockReleaseBuffer(fc_bucketbuf);
}

/*
 * 重放哈希索引的挤压页面操作
 */
static void fc_hash_xlog_squeeze_page(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_squeeze_page *fc_xldata = (xl_hash_squeeze_page *) XLogRecGetData(fc_record);
	Buffer		fc_bucketbuf = InvalidBuffer;
	Buffer		fc_writebuf;
	Buffer		fc_ovflbuf;
	Buffer		fc_prevbuf = InvalidBuffer;
	Buffer		fc_mapbuf;
	XLogRedoAction fc_action;

	/*
	 * 在开始实际重放操作之前，确保我们在主桶页面上有一个清理锁。
	 * 这样可以确保在此操作的重放期间既没有扫描可以开始，也没有扫描正在进行。
	 * 如果我们在此操作期间允许扫描，则它们可能会漏掉某些记录，
	 * 或者重复显示同一条记录。
	 */
	if (fc_xldata->is_prim_bucket_same_wrt)
		fc_action = XLogReadBufferForRedoExtended(fc_record, 1, RBM_NORMAL, true, &fc_writebuf);
	else
	{
		/*
		 * 我们不在乎返回值，因为读取bucketbuf的目的是确保在主桶页面上有一个清理锁。
		 */
		(void) XLogReadBufferForRedoExtended(fc_record, 0, RBM_NORMAL, true, &fc_bucketbuf);

		fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_writebuf);
	}

	/* 重放用于在溢出缓冲区中添加条目的记录 */
	if (fc_action == BLK_NEEDS_REDO)
	{
		Page		fc_writepage;
		char	   *fc_begin;
		char	   *fc_data;
		Size		fc_datalen;
		uint16		fc_ninserted = 0;

		fc_data = fc_begin = XLogRecGetBlockData(fc_record, 1, &fc_datalen);

		fc_writepage = (Page) BufferGetPage(fc_writebuf);

		if (fc_xldata->ntups > 0)
		{
			OffsetNumber *fc_towrite = (OffsetNumber *) fc_data;

			fc_data += sizeof(OffsetNumber) * fc_xldata->ntups;

			while (fc_data - fc_begin < fc_datalen)
			{
				IndexTuple	fc_itup = (IndexTuple) fc_data;
				Size		fc_itemsz;
				OffsetNumber fc_l;

				fc_itemsz = IndexTupleSize(fc_itup);
				fc_itemsz = MAXALIGN(fc_itemsz);

				fc_data += fc_itemsz;

				fc_l = PageAddItem(fc_writepage, (Item) fc_itup, fc_itemsz, fc_towrite[fc_ninserted], false, false);
				if (fc_l == InvalidOffsetNumber)
					elog(ERROR, "hash_xlog_squeeze_page: failed to add item to hash index page, size %d bytes",
						 (int) fc_itemsz);

				fc_ninserted++;
			}
		}

		/*
		 * 插入的元组数量必须与REDO记录中请求的相同。
		 */
		Assert(fc_ninserted == fc_xldata->ntups);

		/*
		 * 如果添加元组的页面在已释放的溢出页面之前，则更新其nextblkno。
		 */
		if (fc_xldata->is_prev_bucket_same_wrt)
		{
			HashPageOpaque fc_writeopaque = HashPageGetOpaque(fc_writepage);

			fc_writeopaque->hasho_nextblkno = fc_xldata->nextblkno;
		}

		PageSetLSN(fc_writepage, fc_lsn);
		MarkBufferDirty(fc_writebuf);
	}

	/* 重放用于初始化溢出缓冲区的记录 */
	if (XLogReadBufferForRedo(fc_record, 2, &fc_ovflbuf) == BLK_NEEDS_REDO)
	{
		Page		fc_ovflpage;
		HashPageOpaque fc_ovflopaque;

		fc_ovflpage = BufferGetPage(fc_ovflbuf);

		_hash_pageinit(fc_ovflpage, BufferGetPageSize(fc_ovflbuf));

		fc_ovflopaque = HashPageGetOpaque(fc_ovflpage);

		fc_ovflopaque->hasho_prevblkno = InvalidBlockNumber;
		fc_ovflopaque->hasho_nextblkno = InvalidBlockNumber;
		fc_ovflopaque->hasho_bucket = InvalidBucket;
		fc_ovflopaque->hasho_flag = LH_UNUSED_PAGE;
		fc_ovflopaque->hasho_page_id = HASHO_PAGE_ID;

		PageSetLSN(fc_ovflpage, fc_lsn);
		MarkBufferDirty(fc_ovflbuf);
	}
	if (BufferIsValid(fc_ovflbuf))
		UnlockReleaseBuffer(fc_ovflbuf);

	/* 重放用于已释放溢出页面之前的页面的记录 */
	if (!fc_xldata->is_prev_bucket_same_wrt &&
		XLogReadBufferForRedo(fc_record, 3, &fc_prevbuf) == BLK_NEEDS_REDO)
	{
		Page		fc_prevpage = BufferGetPage(fc_prevbuf);
		HashPageOpaque fc_prevopaque = HashPageGetOpaque(fc_prevpage);

		fc_prevopaque->hasho_nextblkno = fc_xldata->nextblkno;

		PageSetLSN(fc_prevpage, fc_lsn);
		MarkBufferDirty(fc_prevbuf);
	}
	if (BufferIsValid(fc_prevbuf))
		UnlockReleaseBuffer(fc_prevbuf);

	/* 重放用于已释放溢出页面之后的页面的记录 */
	if (XLogRecHasBlockRef(fc_record, 4))
	{
		Buffer		fc_nextbuf;

		if (XLogReadBufferForRedo(fc_record, 4, &fc_nextbuf) == BLK_NEEDS_REDO)
		{
			Page		fc_nextpage = BufferGetPage(fc_nextbuf);
			HashPageOpaque fc_nextopaque = HashPageGetOpaque(fc_nextpage);

			fc_nextopaque->hasho_prevblkno = fc_xldata->prevblkno;

			PageSetLSN(fc_nextpage, fc_lsn);
			MarkBufferDirty(fc_nextbuf);
		}
		if (BufferIsValid(fc_nextbuf))
			UnlockReleaseBuffer(fc_nextbuf);
	}

	if (BufferIsValid(fc_writebuf))
		UnlockReleaseBuffer(fc_writebuf);

	if (BufferIsValid(fc_bucketbuf))
		UnlockReleaseBuffer(fc_bucketbuf);

	/*
	 * 注意：在正常操作中，我们会在仍持有对主桶页面和溢出页面的锁时更新位图和元数据页面。
	 * 但在重放期间，没必要保持这些锁，因为没有其他索引更新可以同时发生。
	 */
	/* 重放位图页面的记录 */
	if (XLogReadBufferForRedo(fc_record, 5, &fc_mapbuf) == BLK_NEEDS_REDO)
	{
		Page		fc_mappage = (Page) BufferGetPage(fc_mapbuf);
		uint32	   *fc_freep = NULL;
		char	   *fc_data;
		uint32	   *fc_bitmap_page_bit;
		Size		fc_datalen;

		fc_freep = HashPageGetBitmap(fc_mappage);

		fc_data = XLogRecGetBlockData(fc_record, 5, &fc_datalen);
		fc_bitmap_page_bit = (uint32 *) fc_data;

		CLRBIT(fc_freep, *fc_bitmap_page_bit);

		PageSetLSN(fc_mappage, fc_lsn);
		MarkBufferDirty(fc_mapbuf);
	}
	if (BufferIsValid(fc_mapbuf))
		UnlockReleaseBuffer(fc_mapbuf);

	/* 重放元页面的记录 */
	if (XLogRecHasBlockRef(fc_record, 6))
	{
		Buffer		fc_metabuf;

		if (XLogReadBufferForRedo(fc_record, 6, &fc_metabuf) == BLK_NEEDS_REDO)
		{
			HashMetaPage fc_metap;
			Page		fc_page;
			char	   *fc_data;
			uint32	   *fc_firstfree_ovflpage;
			Size		fc_datalen;

			fc_data = XLogRecGetBlockData(fc_record, 6, &fc_datalen);
			fc_firstfree_ovflpage = (uint32 *) fc_data;

			fc_page = BufferGetPage(fc_metabuf);
			fc_metap = HashPageGetMeta(fc_page);
			fc_metap->hashm_firstfree = *fc_firstfree_ovflpage;

			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_metabuf);
		}
		if (BufferIsValid(fc_metabuf))
			UnlockReleaseBuffer(fc_metabuf);
	}
}

/*
 * 重放哈希索引的删除操作
 */
static void fc_hash_xlog_delete(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_delete *fc_xldata = (xl_hash_delete *) XLogRecGetData(fc_record);
	Buffer		fc_bucketbuf = InvalidBuffer;
	Buffer		fc_deletebuf;
	Page		fc_page;
	XLogRedoAction fc_action;

	/*
	 * 在开始实际重放操作之前，确保我们在主桶页面上有一个清理锁。
	 * 这样可以确保在此操作的重放期间既没有扫描可以开始，也没有扫描正在进行。
	 * 如果我们在此操作期间允许扫描，则它们可能会漏掉某些记录，
	 * 或者重复显示同一条记录。
	 */
	if (fc_xldata->is_primary_bucket_page)
		fc_action = XLogReadBufferForRedoExtended(fc_record, 1, RBM_NORMAL, true, &fc_deletebuf);
	else
	{
		/*
		 * 我们不在乎返回值，因为读取bucketbuf的目的是确保在主桶页面上有一个清理锁。
		 */
		(void) XLogReadBufferForRedoExtended(fc_record, 0, RBM_NORMAL, true, &fc_bucketbuf);

		fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_deletebuf);
	}

	/* 重放删除桶页面中条目的记录 */
	if (fc_action == BLK_NEEDS_REDO)
	{
		char	   *fc_ptr;
		Size		fc_len;

		fc_ptr = XLogRecGetBlockData(fc_record, 1, &fc_len);

		fc_page = (Page) BufferGetPage(fc_deletebuf);

		if (fc_len > 0)
		{
			OffsetNumber *fc_unused;
			OffsetNumber *fc_unend;

			fc_unused = (OffsetNumber *) fc_ptr;
			fc_unend = (OffsetNumber *) ((char *) fc_ptr + fc_len);

			if ((fc_unend - fc_unused) > 0)
				PageIndexMultiDelete(fc_page, fc_unused, fc_unend - fc_unused);
		}

		/*
		 * 只有在 clear_dead_marking 标志设置为 true 的情况下，
		 * 才将页面标记为不包含任何 LP_DEAD 项。有关详细信息，请参见 hashbucketcleanup() 中的注释。
		 */
		if (fc_xldata->clear_dead_marking)
		{
			HashPageOpaque fc_pageopaque;

			fc_pageopaque = HashPageGetOpaque(fc_page);
			fc_pageopaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_deletebuf);
	}
	if (BufferIsValid(fc_deletebuf))
		UnlockReleaseBuffer(fc_deletebuf);

	if (BufferIsValid(fc_bucketbuf))
		UnlockReleaseBuffer(fc_bucketbuf);
}

/*
 * 重放主桶页面的分裂清理标志操作。
 */
static void fc_hash_xlog_split_cleanup(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	Buffer		fc_buffer;
	Page		fc_page;

	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		HashPageOpaque fc_bucket_opaque;

		fc_page = (Page) BufferGetPage(fc_buffer);

		fc_bucket_opaque = HashPageGetOpaque(fc_page);
		fc_bucket_opaque->hasho_flag &= ~LH_BUCKET_NEEDS_SPLIT_CLEANUP;
		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

/*
 * 重放更新元页面
 */
static void fc_hash_xlog_update_meta_page(XLogReaderState *fc_record)
{
	HashMetaPage fc_metap;
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_update_meta_page *fc_xldata = (xl_hash_update_meta_page *) XLogRecGetData(fc_record);
	Buffer		fc_metabuf;
	Page		fc_page;

	if (XLogReadBufferForRedo(fc_record, 0, &fc_metabuf) == BLK_NEEDS_REDO)
	{
		fc_page = BufferGetPage(fc_metabuf);
		fc_metap = HashPageGetMeta(fc_page);

		fc_metap->hashm_ntuples = fc_xldata->ntuples;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_metabuf);
	}
	if (BufferIsValid(fc_metabuf))
		UnlockReleaseBuffer(fc_metabuf);
}

/*
 * 重放哈希索引中的删除操作，以移除在索引元组插入过程中标记为 DEAD 的元组。
 */
static void fc_hash_xlog_vacuum_one_page(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_hash_vacuum_one_page *fc_xldata;
	Buffer		fc_buffer;
	Buffer		fc_metabuf;
	Page		fc_page;
	XLogRedoAction fc_action;
	HashPageOpaque fc_pageopaque;

	fc_xldata = (xl_hash_vacuum_one_page *) XLogRecGetData(fc_record);

	/*
	 * 如果我们有任何冲突处理，则必须在更新页面之前进行。
	 *
	 * 在哈希索引元组插入期间被标记为 LP_DEAD 的哈希索引记录
	 * 可能会与备用查询发生冲突。您可能会认为垃圾回收记录也会发生冲突，
	 * 但我们已经处理过了。XLOG_HEAP2_PRUNE 记录提供通过堆的垃圾回收清理的最高 xid，
	 * 因此我们可以在它到达时解决任何冲突。之后，我们知道该索引上的
	 * 单个哈希索引垃圾回收记录不存在冲突。
	 */
	if (InHotStandby)
	{
		RelFileNode fc_rnode;

		XLogRecGetBlockTag(fc_record, 0, &fc_rnode, NULL, NULL);
		ResolveRecoveryConflictWithSnapshot(fc_xldata->latestRemovedXid, fc_rnode);
	}

	fc_action = XLogReadBufferForRedoExtended(fc_record, 0, RBM_NORMAL, true, &fc_buffer);

	if (fc_action == BLK_NEEDS_REDO)
	{
		fc_page = (Page) BufferGetPage(fc_buffer);

		if (XLogRecGetDataLen(fc_record) > SizeOfHashVacuumOnePage)
		{
			OffsetNumber *fc_unused;

			fc_unused = (OffsetNumber *) ((char *) fc_xldata + SizeOfHashVacuumOnePage);

			PageIndexMultiDelete(fc_page, fc_unused, fc_xldata->ntuples);
		}

		/*
		 * 将页面标记为不包含任何 LP_DEAD 项。有关详细信息，请参见
		 * _hash_vacuum_one_page() 中的注释。
		 */
		fc_pageopaque = HashPageGetOpaque(fc_page);
		fc_pageopaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	if (XLogReadBufferForRedo(fc_record, 1, &fc_metabuf) == BLK_NEEDS_REDO)
	{
		Page		fc_metapage;
		HashMetaPage fc_metap;

		fc_metapage = BufferGetPage(fc_metabuf);
		fc_metap = HashPageGetMeta(fc_metapage);

		fc_metap->hashm_ntuples -= fc_xldata->ntuples;

		PageSetLSN(fc_metapage, fc_lsn);
		MarkBufferDirty(fc_metabuf);
	}
	if (BufferIsValid(fc_metabuf))
		UnlockReleaseBuffer(fc_metabuf);
}

void hash_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;

	switch (fc_info)
	{
		case XLOG_HASH_INIT_META_PAGE:
			fc_hash_xlog_init_meta_page(fc_record);
			break;
		case XLOG_HASH_INIT_BITMAP_PAGE:
			fc_hash_xlog_init_bitmap_page(fc_record);
			break;
		case XLOG_HASH_INSERT:
			fc_hash_xlog_insert(fc_record);
			break;
		case XLOG_HASH_ADD_OVFL_PAGE:
			fc_hash_xlog_add_ovfl_page(fc_record);
			break;
		case XLOG_HASH_SPLIT_ALLOCATE_PAGE:
			fc_hash_xlog_split_allocate_page(fc_record);
			break;
		case XLOG_HASH_SPLIT_PAGE:
			fc_hash_xlog_split_page(fc_record);
			break;
		case XLOG_HASH_SPLIT_COMPLETE:
			fc_hash_xlog_split_complete(fc_record);
			break;
		case XLOG_HASH_MOVE_PAGE_CONTENTS:
			fc_hash_xlog_move_page_contents(fc_record);
			break;
		case XLOG_HASH_SQUEEZE_PAGE:
			fc_hash_xlog_squeeze_page(fc_record);
			break;
		case XLOG_HASH_DELETE:
			fc_hash_xlog_delete(fc_record);
			break;
		case XLOG_HASH_SPLIT_CLEANUP:
			fc_hash_xlog_split_cleanup(fc_record);
			break;
		case XLOG_HASH_UPDATE_META_PAGE:
			fc_hash_xlog_update_meta_page(fc_record);
			break;
		case XLOG_HASH_VACUUM_ONE_PAGE:
			fc_hash_xlog_vacuum_one_page(fc_record);
			break;
		default:
			elog(PANIC, "hash_redo: unknown op code %u", fc_info);
	}
}

/*
 * 在对哈希页面执行一致性检查之前，进行掩码处理。
 */
void hash_mask(char *fc_pagedata, BlockNumber fc_blkno)
{
	Page		fc_page = (Page) fc_pagedata;
	HashPageOpaque fc_opaque;
	int			fc_pagetype;

	mask_page_lsn_and_checksum(fc_page);

	mask_page_hint_bits(fc_page);
	mask_unused_space(fc_page);

	fc_opaque = HashPageGetOpaque(fc_page);

	fc_pagetype = fc_opaque->hasho_flag & LH_PAGE_TYPE;
	if (fc_pagetype == LH_UNUSED_PAGE)
	{
		/*
		 * 在 UNUSED 页面上掩码所有内容。
		 */
		mask_page_content(fc_page);
	}
	else if (fc_pagetype == LH_BUCKET_PAGE ||
			 fc_pagetype == LH_OVERFLOW_PAGE)
	{
		/*
		 * 在哈希桶和溢出页面中，有可能修改 LP_FLAGS 而不产生任何 WAL 记录。
		 * 因此，掩码行指针标志。有关详细信息，请参见 hashgettuple()、_hash_kill_items()。
		 */
		mask_lp_flags(fc_page);
	}

	/*
	 * 提示位 LH_PAGE_HAS_DEAD_TUPLES 可能保持未记录状态。
	 * 因此，进行掩码处理。有关详细信息，请参见 _hash_kill_items()。
	 */
	fc_opaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;
}
